package com.yongche.kafka

import org.apache.curator.framework.CuratorFrameworkFactory
import org.apache.curator.retry.ExponentialBackoffRetry

/**
  * Created by yd on 2016/10/10.
  */
case class OffsetStore(connectionString: String) {

  def getZKClient() = {
    //创建zookeeper的客户端
    val c = CuratorFrameworkFactory.builder()
      .connectString(connectionString)
      .retryPolicy(new ExponentialBackoffRetry(1000, 3))
      .connectionTimeoutMs(1000)
      .sessionTimeoutMs(1000)
      .build()
    c.start()
    c
  }


  //保存path的值，如果不存在，则创建，如果存在，则更新
  def save(path: String, offsetJson: String): Unit = {
    val client = getZKClient()
    val stat = client.checkExists().forPath(path)
    if (stat == null) {
      client.create().creatingParentsIfNeeded().forPath(path, offsetJson.getBytes)
    } else {
      client.setData().forPath(path, offsetJson.getBytes);
    }
    client.close()
  }

  //查询指定路径的值
  def query(path: String): String = {
    val client = getZKClient()
    val stat = client.checkExists().forPath(path)
    val ret = stat match {
      case null => null
      case _ => {
        val bytes = client.getData.forPath(path)
        new String(bytes)
      }
    }
    client.close()
    ret
  }

}
